package belf.migrate.engine.admin.handler;

import belf.migrate.api.common.DbConstant;
import belf.migrate.api.exception.CatalogException;
import belf.migrate.api.exception.DatabaseNotExistException;
import belf.migrate.api.exception.TableNotExistException;
import belf.migrate.api.model.ConvertSqlModel;
import belf.migrate.api.model.ExecuteSourceBatchSql;
import belf.migrate.api.model.ExecuteSqlModel;
import belf.migrate.api.model.QualifiersModel;
import belf.migrate.api.model.result.BatchExecuteResult;
import belf.migrate.api.table.schema.AbstractJdbcCatalog;

import belf.migrate.api.taskconf.DbObjectConf;
import belf.migrate.api.taskconf.SourceConf;

import belf.migrate.api.table.schema.ColumnMetaData;
import belf.migrate.api.table.schema.TablePath;
import belf.migrate.api.taskconf.*;

import belf.migrate.api.type.DbObjectType;
import belf.migrate.api.type.MigrationObjectType;
import belf.migrate.api.util.DatabaseType;
import belf.migrate.api.util.FieldCase;
import belf.migrate.api.util.QuoteType;
import belf.migrate.engine.ddl.BaseDDLConvert;
import belf.migrate.engine.ddl.function.FunctionConvertFactory;
import belf.migrate.engine.ddl.procedure.ProcedureConvertFactory;
import belf.migrate.engine.ddl.seq.SeqConvertFactory;
import belf.migrate.engine.ddl.synonym.SynonymConvertFactory;
import belf.migrate.engine.ddl.trigger.TriggerConvertFactory;
import belf.migrate.engine.ddl.type.TypeConvertFactory;
import belf.migrate.engine.ddl.view.ViewConvetFactory;
import belf.migrate.engine.execute.ExecuteHandler;
import belf.migrate.engine.job.CatalogCreator;
import belf.migrate.engine.job.Result;

import belf.migrate.engine.job.model.sql.ColumnResult;

import belf.migrate.engine.job.model.sql.DatabaseAndSchema;
import belf.migrate.engine.verify.VerifyHandler;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import spark.Request;
import spark.Response;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;

import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import java.util.*;
import java.util.stream.Collectors;


import static spark.Spark.post;

@Slf4j
public class SqlHandler {


    private static Gson gson = new Gson();

    public static void setupRoutes() {

        /*
          获取database列表接口
         */
        post("/listDatabase", SqlHandler::listDatabase, gson::toJson);

        /*
          获取table列表接口
         */
        post("/listTables", SqlHandler::listTables, gson::toJson);

        /*
          获取object列表接口
         */
        post("/listObjects", SqlHandler::listObjects, gson::toJson);

        /*
          执行sql接口
         */
        post("/executeSql", SqlHandler::executeSql, gson::toJson);

        /*
          批量执行源端sql接口
         */
        post("/batchExecuteSql", SqlHandler::batchExecuteSql, gson::toJson);

        /*
          校验接口
         */
        post("/verify", VerifyHandler::verify, gson::toJson);

        /*
          获取column列表接口
         */
        post("/listColumns", SqlHandler::listColumns, gson::toJson);

        /*
          获取database下所有column列表接口
         */
        post("/listDbColumns", SqlHandler::listDbColumns, gson::toJson);

        /*
          转化sql接口
         */
        post("/convertSql", SqlHandler::convertSql, gson::toJson);

        /*
          获取database和schema列表接口
         */
        post("/listDatabaseAndSchema", SqlHandler::listDatabaseAndSchema, gson::toJson);


    }

    public static JsonObject listDatabase(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        try (AbstractJdbcCatalog catalog = getAbstractJdbcCatalog(request, mapper)) {
            if (catalog == null) {
                return gson.toJsonTree(Result.sqlFail("未查到database")).getAsJsonObject();
            }
            return gson.toJsonTree(Result.success(catalog.listDatabases())).getAsJsonObject();

        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误");
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("message", "解析请求的json格式失败");
            return jsonResponse;
        } catch (CatalogException e) {
            return gson.toJsonTree(Result.sqlFail(e.getMessage())).getAsJsonObject();
        }
    }

    public static JsonObject listDatabaseAndSchema(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        AbstractJdbcCatalog catalog = null;
        try {
            SourceConf sourceConf = mapper.readValue(request.body(), SourceConf.class);
            catalog = CatalogCreator.createCatalog(
                    sourceConf.getCatalogName(),
                    sourceConf.getDatabaseName(),
                    sourceConf.getSchemaName(),
                    sourceConf.getConnectionConf(),
                    sourceConf.isCaseSensitive());
            if (catalog == null) {
                return gson.toJsonTree(Result.sqlFail("未查到database")).getAsJsonObject();
            }
            List<DatabaseAndSchema> result = new ArrayList<>();
            List<String> catalogs = null;
            try {
                catalogs = catalog.listCatalogs();
            } catch (CatalogException e) {
                log.info(sourceConf.getCatalogName() + "不支持获取catalog三层架构");
            }
            if (catalogs == null) {
                List<String> databases = catalog.listDatabases();
                for (String database : databases) {
                    int sort = 1;
                    if (database.equalsIgnoreCase(StringUtils.isNotEmpty(sourceConf.getDatabaseName()) ? sourceConf.getDatabaseName() : sourceConf.getSchemaName())) {
                        sort = 0;
                    }
                    result.add(new DatabaseAndSchema(database, null, sort));
                }
            } else {
                for (String catalogName : catalogs) {
                    int sort = 1;
                    if (catalogName.equalsIgnoreCase(StringUtils.isNotEmpty(sourceConf.getDatabaseName()) ? sourceConf.getDatabaseName() : sourceConf.getSchemaName())) {
                        sort = 0;
                    }
                    List<String> schemas = catalog.listSchemasByDatabase(catalogName);
                    int schemaIndex = schemas.indexOf(sourceConf.getSchemaName());
                    if (schemaIndex != -1) {
                        String element = schemas.remove(schemaIndex);
                        schemas.add(0, element);
                    }
                    result.add(new DatabaseAndSchema(catalogName, schemas, sort));
                }
            }

            result.sort(Comparator.comparingInt(DatabaseAndSchema::getSort));
            return gson.toJsonTree(Result.success(result)).getAsJsonObject();

        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误");
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("message", "解析请求的json格式失败");
            return jsonResponse;
        } catch (CatalogException e) {
            return gson.toJsonTree(Result.sqlFail(e.getMessage())).getAsJsonObject();
        } finally {
            if (catalog != null) {
                try {
                    catalog.close();
                } catch (CatalogException e) {
                    log.error("关闭catalog失败");
                }
            }
        }
    }

    private static AbstractJdbcCatalog getAbstractJdbcCatalog(Request request, ObjectMapper mapper) throws JsonProcessingException {
        SourceConf sourceConf = mapper.readValue(request.body(), SourceConf.class);
        return CatalogCreator.createCatalog(
                sourceConf.getCatalogName(),
                sourceConf.getDatabaseName(),
                sourceConf.getSchemaName(), sourceConf.getConnectionConf(), sourceConf.isCaseSensitive());
    }

    public static JsonObject listObjects(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        DbObjectConf dbObjectConf = null;
        try {
            dbObjectConf = mapper.readValue(request.body(), DbObjectConf.class);
        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误");
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("message", "解析请求的json格式失败");
            return jsonResponse;
        }
        try (AbstractJdbcCatalog catalog = CatalogCreator.createCatalog(
                dbObjectConf.getCatalogName(),
                dbObjectConf.getDatabaseName(),
                dbObjectConf.getSchemaName(),
                dbObjectConf.getConnectionConf(),
                dbObjectConf.isCaseSensitive())) {

            if (catalog == null) {
                return gson.toJsonTree(Result.sqlFail("未查到database")).getAsJsonObject();
            }
            List<String> objectTypeList = Arrays.stream(dbObjectConf.getTaskMode().split(","))
                    .map(String::trim)
                    .toList();
            Map<String, Object> result = new LinkedHashMap<>();
            for (String objectType : objectTypeList) {
                if (objectType.equalsIgnoreCase(DbObjectType.TABLE.getTypeName())) {
                    result.put(DbObjectType.TABLE.getTypeName(), catalog.listTablesInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                } else if (objectType.equalsIgnoreCase(DbObjectType.VIEW.getTypeName())) {
                    result.put(DbObjectType.VIEW.getTypeName(), catalog.listViewsInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                } else if (objectType.equalsIgnoreCase(DbObjectType.PROCEDURE.getTypeName())) {
                    result.put(DbObjectType.PROCEDURE.getTypeName(), catalog.listProceduresInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                } else if (objectType.equalsIgnoreCase(DbObjectType.TYPE.getTypeName())) {
                    result.put(DbObjectType.TYPE.getTypeName(), catalog.listTypesInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                } else if (objectType.equalsIgnoreCase(DbObjectType.TRIGGER.getTypeName())) {
                    result.put(DbObjectType.TRIGGER.getTypeName(), catalog.listTriggersInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                } else if (objectType.equalsIgnoreCase(DbObjectType.ALL.getTypeName())) {
                    result.put(DbObjectType.TABLE.getTypeName(), catalog.listTablesInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                    result.put(DbObjectType.VIEW.getTypeName(), catalog.listViewsInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                    result.put(DbObjectType.PROCEDURE.getTypeName(), catalog.listProceduresInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                    result.put(DbObjectType.TYPE.getTypeName(), catalog.listTypesInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                    result.put(DbObjectType.TRIGGER.getTypeName(), catalog.listTriggersInfo(dbObjectConf.getDatabaseName(), dbObjectConf.getSchemaName()));
                }
            }
            return gson.toJsonTree(Result.success(result)).getAsJsonObject();

        } catch (CatalogException | DatabaseNotExistException e) {
            return gson.toJsonTree(Result.sqlFail(e.getMessage())).getAsJsonObject();
        }
    }

    public static JsonObject listTables(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        SourceConf sourceConf = null;
        try {
            sourceConf = mapper.readValue(request.body(), SourceConf.class);
        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误");
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("message", "解析请求的json格式失败");
            return jsonResponse;
        }
        try (AbstractJdbcCatalog catalog = CatalogCreator.createCatalog(
                sourceConf.getCatalogName(),
                sourceConf.getDatabaseName(),
                sourceConf.getSchemaName(), sourceConf.getConnectionConf(), sourceConf.isCaseSensitive())) {

            if (catalog == null) {
                return gson.toJsonTree(Result.sqlFail("未查到database")).getAsJsonObject();
            }
            return gson.toJsonTree(Result.success(catalog.listTables(sourceConf.getDatabaseName(), sourceConf.getSchemaName()))).getAsJsonObject();

        } catch (CatalogException | DatabaseNotExistException e) {
            return gson.toJsonTree(Result.sqlFail(e.getMessage())).getAsJsonObject();
        }
    }

    public static JsonObject listColumns(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        ColumnRequestConf columnRequestConf = null;
        try {
            columnRequestConf = mapper.readValue(request.body(), ColumnRequestConf.class);
        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误");
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("message", "解析请求的json格式失败");
            return jsonResponse;
        }
        try (AbstractJdbcCatalog catalog = CatalogCreator.createCatalog(
                columnRequestConf.getCatalogName(),
                columnRequestConf.getDatabaseName(),
                columnRequestConf.getSchemaName(), columnRequestConf.getConnectionConf(), columnRequestConf.isCaseSensitive())) {

            if (catalog == null) {
                return gson.toJsonTree(Result.sqlFail("未查到database")).getAsJsonObject();
            }
            FieldCase fieldCase = FieldCase.getFieldCase(columnRequestConf.getFieldCase());
            DatabaseType databaseType = DatabaseType.getDatabaseType(columnRequestConf.getCatalogName());

            String leftQuote = columnRequestConf.isQuote() ? databaseType.getQuoteType().getValue() : QuoteType.NONE.getValue();
            String rightQuote = leftQuote;
            if (leftQuote.length() == 2) {
                rightQuote = leftQuote.substring(1);
                leftQuote = leftQuote.substring(0, 1);
            }

            QualifiersModel qualifiersModel = new QualifiersModel(leftQuote, rightQuote, fieldCase, databaseType.getQualifiedType());
            TablePath tablePath = TablePath.of(columnRequestConf.getDatabaseName(), columnRequestConf.getSchemaName(), columnRequestConf.getTableName(), qualifiersModel);
            List<ColumnMetaData> columns = catalog.getAllColumnsBySql(tablePath);
            List<ColumnResult> columnResults = getColumnResults(columns);

            columnResults.sort(Comparator.comparingInt(ColumnResult::getSort));
            return gson.toJsonTree(Result.success(columnResults)).getAsJsonObject();

        } catch (CatalogException | TableNotExistException e) {
            return gson.toJsonTree(Result.sqlFail(e.getMessage())).getAsJsonObject();
        }
    }

    public static JsonObject listDbColumns(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        DbColumnsConf columnsConf = null;
        try {
            columnsConf = mapper.readValue(request.body(), DbColumnsConf.class);
        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误");
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("message", "解析请求的json格式失败");
            return jsonResponse;
        }
        try (AbstractJdbcCatalog catalog = CatalogCreator.createCatalog(
                columnsConf.getCatalogName(),
                columnsConf.getDatabaseName(),
                columnsConf.getSchemaName(), columnsConf.getConnectionConf(), columnsConf.isCaseSensitive())) {

            if (catalog == null) {
                return gson.toJsonTree(Result.sqlFail("未查到database")).getAsJsonObject();
            }
            FieldCase fieldCase = FieldCase.getFieldCase(columnsConf.getFieldCase());
            DatabaseType databaseType = DatabaseType.getDatabaseType(columnsConf.getCatalogName());

            String leftQuote = columnsConf.isQuote() ? databaseType.getQuoteType().getValue() : QuoteType.NONE.getValue();
            String rightQuote = leftQuote;
            if (leftQuote.length() == 2) {
                rightQuote = leftQuote.substring(1);
                leftQuote = leftQuote.substring(0, 1);
            }
            QualifiersModel qualifiersModel = new QualifiersModel(leftQuote, rightQuote, fieldCase, databaseType.getQualifiedType());
            String tableName = String.join("','", columnsConf.getTables());
            TablePath tablePath = TablePath.of(columnsConf.getDatabaseName(), columnsConf.getSchemaName(), tableName, qualifiersModel);
            List<ColumnMetaData> columns = catalog.getAllColumnsBySql(tablePath);

            Map<String, List<ColumnResult>> columnMap = new HashMap<>();
            for (ColumnMetaData column : columns) {
                if (columnMap.containsKey(column.getTableName())) {
                    columnMap.get(column.getTableName()).add(new ColumnResult(column));
                } else {
                    List<ColumnResult> columnResults = new ArrayList<>();
                    columnResults.add(new ColumnResult(column));
                    columnMap.put(column.getTableName(), columnResults);
                }
            }

            Map<String, List<String>> result = new HashMap<>();
            for (String key : columnMap.keySet()) {
                List<ColumnResult> columnResults = columnMap.get(key);
                columnResults.sort(Comparator.comparingInt(ColumnResult::getSort));
                result.put(key, columnResults.stream().map(ColumnResult::getColumnName).collect(Collectors.toList()));
            }

            return gson.toJsonTree(Result.success(result)).getAsJsonObject();

        } catch (CatalogException | TableNotExistException e) {
            return gson.toJsonTree(Result.sqlFail(e.getMessage())).getAsJsonObject();
        }
    }

    private static List<ColumnResult> getColumnResults(List<ColumnMetaData> columns) {
        List<ColumnResult> columnResults = new ArrayList<>();
        for (ColumnMetaData column : columns) {
            ColumnResult columnResult = new ColumnResult();
            columnResult.setColumnName(column.getColumnName());
            columnResult.setColumnType(column.getColumnTypeName());
            columnResult.setComment(column.getColumnComment());
            columnResult.setTableName(column.getTableName());
            if (column.isPrimaryKey()) {
                columnResult.setSort(0);
            } else if (DbConstant.DATETIME_TYPES.contains(column.getColumnTypeName().toLowerCase(Locale.ROOT))) {
                columnResult.setSort(1);
            } else {
                columnResult.setSort(2);
            }
            columnResults.add(columnResult);
        }
        return columnResults;
    }

    public static JsonObject executeSql(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        Connection connection = null;
        long startTime = System.currentTimeMillis();
        try {
            ExecuteSqlModel executeSqlModel = mapper.readValue(request.body(), ExecuteSqlModel.class);
            AbstractJdbcCatalog catalog = CatalogCreator.createCatalog(
                    executeSqlModel.getCatalogName(),
                    executeSqlModel.getDatabaseName(),
                    executeSqlModel.getSchemaName(), executeSqlModel.getConnectionConf(), executeSqlModel.isCaseSensitive());
            connection = DriverManager.getConnection(executeSqlModel.getConnectionConf().getUrl(), executeSqlModel.getConnectionConf().getUser(), executeSqlModel.getConnectionConf().getPassword());
            //设置当前数据库和当前schema
            try {
                catalog.setCurrentDatabase(connection, executeSqlModel.getDatabaseName());
                catalog.setCurrentSchema(connection, executeSqlModel.getSchemaName());
            } catch (SQLException e) {
                log.error("设置当前数据库失败", e);
            }
            String sql = catalog.trimSql(executeSqlModel.getSql());
            ExecuteHandler.executeDDL(connection, sql);
            log.info("sql执行成功！Sql = " + executeSqlModel.getSql());
        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误", e);
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("msg", "解析请求的json格式失败");
            return jsonResponse;
        } catch (SQLException e) {
            log.error("执行sql失败", e);
            return gson.toJsonTree(Result.fail(e.getMessage())).getAsJsonObject();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    log.error("关闭数据库连接失败", e);
                }
            }
        }
        long endTime = System.currentTimeMillis();
        String message = String.format("执行成功, 执行耗时%s毫秒.", endTime - startTime);
        return gson.toJsonTree(Result.success(null, message)).getAsJsonObject();
    }

    public static JsonObject batchExecuteSql(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        Connection connection = null;
        long startTime = System.currentTimeMillis();
        List<BatchExecuteResult> results = null;
        try {
            ExecuteSourceBatchSql batchSql = mapper.readValue(request.body(), ExecuteSourceBatchSql.class);
            AbstractJdbcCatalog catalog = CatalogCreator.createCatalog(
                    batchSql.getCatalogName(),
                    batchSql.getDatabaseName(),
                    batchSql.getSchemaName(), batchSql.getConnectionConf(), batchSql.isCaseSensitive());
            connection = DriverManager.getConnection(batchSql.getConnectionConf().getUrl(), batchSql.getConnectionConf().getUser(), batchSql.getConnectionConf().getPassword());
            //设置当前数据库和当前schema
            try {
                catalog.setCurrentDatabase(connection, batchSql.getDatabaseName());
                catalog.setCurrentSchema(connection, batchSql.getSchemaName());
            } catch (SQLException e) {
                log.error("设置当前数据库失败", e);
            }
            results = new ArrayList<>();
            for (String sql : batchSql.getSqls()) {
                BatchExecuteResult result = new BatchExecuteResult();
                result.setSql(sql);
                result.setSuccess(true);
                try {
                    ExecuteHandler.executeDDL(connection, sql);
                } catch (SQLException e) {
                    result.setSuccess(false);
                    result.setMessage(e.getMessage());
                }
                results.add(result);
            }
        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误", e);
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("msg", "解析请求的json格式失败");
            return jsonResponse;
        } catch (SQLException e) {
            log.error("批量执行sql，打开数据源失败", e);
            return gson.toJsonTree(Result.fail(e.getMessage())).getAsJsonObject();
        } finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (SQLException e) {
                    log.error("关闭数据库连接失败", e);
                }
            }
        }
        long endTime = System.currentTimeMillis();
        String message = String.format("执行成功, 执行耗时%s毫秒.", endTime - startTime);
        return gson.toJsonTree(Result.success(results, message)).getAsJsonObject();
    }

    public static JsonObject convertSql(Request request, Response response) {
        // 设置响应类型为JSON
        response.type("application/json");
        // 获取请求体中的JSON数据并反序列化为Job对象
        ObjectMapper mapper = new ObjectMapper();
        BaseDDLConvert convert = null;
        try {
            ConvertSqlModel convertSqlModel = mapper.readValue(request.body(), ConvertSqlModel.class);
            DatabaseType databaseType = DatabaseType.getDatabaseType(convertSqlModel.getSinkConf().getCatalogName());
            FieldCase fieldCase = FieldCase.getFieldCase(convertSqlModel.getSinkConf().getFieldCase());

            AbstractJdbcCatalog sourceCatalog = CatalogCreator.createCatalog(
                    convertSqlModel.getSourceConf().getCatalogName(),
                    convertSqlModel.getSourceConf().getDatabaseName(),
                    convertSqlModel.getSourceConf().getSchemaName(),
                    convertSqlModel.getSourceConf().getConnectionConf(),
                    convertSqlModel.getSourceConf().isCaseSensitive());

            AbstractJdbcCatalog sinkCatalog = CatalogCreator.createCatalog(
                    convertSqlModel.getSinkConf().getCatalogName(),
                    convertSqlModel.getSinkConf().getDatabaseName(),
                    convertSqlModel.getSinkConf().getSchemaName(),
                    convertSqlModel.getSinkConf().getConnectionConf(),
                    convertSqlModel.getSinkConf().isCaseSensitive());

            String leftQuote = convertSqlModel.getSinkConf().isQuote() ? databaseType.getQuoteType().getValue() : QuoteType.NONE.getValue();
            String rightQuote = leftQuote;
            if (leftQuote.length() == 2) {
                rightQuote = leftQuote.substring(1);
                leftQuote = leftQuote.substring(0, 1);
            }

            QualifiersModel qualifiersModel = new QualifiersModel(leftQuote, rightQuote, fieldCase, databaseType.getQualifiedType());

            MigrationObjectType migrationObjectType = MigrationObjectType.getMigrationObjectType(convertSqlModel.getType());

            convert = switch (migrationObjectType) {
                case VIEW -> new ViewConvetFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                case FUNCTION -> new FunctionConvertFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                case SYNONYM -> new SynonymConvertFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                case TRIGGER -> new TriggerConvertFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                case TYPE -> new TypeConvertFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                case SEQUENCE -> new SeqConvertFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                case PROCEDURE -> new ProcedureConvertFactory().create(
                        getValidSinkDatabase(convertSqlModel.getSinkConf()),
                        convertSqlModel.getName(),
                        convertSqlModel.getSql(),
                        sourceCatalog,
                        sinkCatalog,
                        qualifiersModel);
                default -> convert;
            };

            String ddl = "";
            if (convert != null) {
                ddl = convert.convert();
            }
            return gson.toJsonTree(Result.success(ddl, "转换成功")).getAsJsonObject();

        } catch (JsonProcessingException e) {
            log.error("解析参数格式错误", e);
            JsonObject jsonResponse = new JsonObject();
            jsonResponse.addProperty("code", 11);
            jsonResponse.addProperty("msg", "解析请求的json格式失败");
            return jsonResponse;
        }
    }

    public static String getValidSinkDatabase(SinkConf sinkConf) {
        if (StringUtils.isNotBlank(sinkConf.getSchemaName())) {
            return sinkConf.getSchemaName();
        }
        return sinkConf.getDatabaseName();
    }
}
